package datastructure.blockingqueue;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 双锁实现阻塞队列
 *
 * @author 杨 强
 * @package datastructure.blockingqueue
 * @createTime 2023-03-18 22:17
 */
public class TwoLockBlockingQueue<E> implements BlockingQueue<E>{
    /**
     * 储存数据的数组
     */
    private final E[] array;
    /**
     * 尾指针,就是该指针的大小就是添加元素应该存储的下标.
     */
    private int head = 0;
    /**
     * 头指针,就是该指针的大小就是取元素的下标.
     */
    private int tail = 0;
    /**
     * 队列元素个数, 用原子类来保护.
     */
    private AtomicInteger size = new AtomicInteger();

    /**
     * 构造器初始化队列容量
     * @param capacity
     */
    public TwoLockBlockingQueue(int capacity) {
        array = (E[]) new Object[capacity];
    }

    /**
     * 头锁
     */
    ReentrantLock headLock = new ReentrantLock();
    Condition headWaits = headLock.newCondition();
    /**
     * 尾锁
     */
    ReentrantLock tailLock = new ReentrantLock();
    Condition tailWaits = tailLock.newCondition();

    @Override
    public void offer(E e) throws InterruptedException {
        // 加可以被打断的锁
        tailLock.lockInterruptibly();
        // 新增前队列元素的个数
        int c;
        try {
            // 判断是否满了,用while是为了唤醒也继续判断一下,防止虚假唤醒
            while (isFull()) {
                // 满了就在条件队列阻塞.
                tailWaits.await();
            }
            // 添加数据
            array[tail] = e;
            // tail指针+1, 并且判断加一后是否等于数组长度,等于的话就又从0开始.
            if (++tail == array.length) {
                tail = 0;
            }
            // 队列元素加一,并且保证线程安全
            // 这个方法是先获取值再新增,也就是c是新增前的值,0代表就是队列没有元素
            c = size.getAndIncrement();
        } finally {
            // 释放锁
            tailLock.unlock();
        }

        // 获取headLock锁, 唤醒阻塞在headWaits条件队列的线程,可以进行取数据了
        if (c == 0){
            headLock.lock();
            try {
                headWaits.signal();
            }finally {
                headLock.unlock();
            }
        }
    }

    @Override
    public void offer(E e, Long timeout) throws InterruptedException {
       return;
    }

    @Override
    public E poll() throws InterruptedException {
        // 加可以被打断的锁
        headLock.lockInterruptibly();
        E e;
        // 取走前队列元的个数
        int c;
        try {
            // 判断是否空,用while是为了唤醒也继续判断一下,防止虚假唤醒
            while (isEmpty()) {
                // 在headWaits条件队列阻塞
                headWaits.await();
            }
            // 取出元素
            e = array[head];
            array[head] = null; // help GC
            if (++head == array.length) {
                head = 0;
            }
            // 原子操作
            c = size.getAndDecrement();
            // 队列元素大于1就唤醒等待取元素的阻塞线程.
            if (c > 1){
                headWaits.signal();
            }
        } finally {
            headLock.unlock();
        }

        // 获取tailLock锁,取走元素后队列不满了,然后通知等待添加元素的线程
        // 队列是满的才唤醒等待添加的线程.
        if (c == array.length){
            tailLock.lock();
            try {
                tailWaits.signal();
            } finally {
                tailLock.unlock();
            }
        }
        return e;
    }

    /**
     * 判空
     * @return
     */
    private boolean isEmpty() {
        return size.get() == 0;
    }

    /**
     * 判满
     * @return
     */
    private boolean isFull() {
        return size.get() == array.length;
    }
}
